package iot.disruptor;

import iot.DataDispatcher;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by dave on 18-8-11 下午9:42.
 */
public class DisruptorDispatcher implements DataDispatcher, InitializingBean, DisposableBean {
    @Autowired(required = false) List<DisruptorQueue> queues;
    private final Map<Class, DisruptorQueue> queueMap = new HashMap<>();
    private ThreadPoolTaskExecutor executor; // 用于写入数据操作的线程池

    public DisruptorDispatcher(int writeThreadCount) {
        executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(writeThreadCount);
        executor.setMaxPoolSize(writeThreadCount * 2);
        executor.setQueueCapacity(writeThreadCount * 10);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (queues != null && queues.size() > 0) {
            queues.forEach(q -> queueMap.put(q.getEntityType(), q));
            executor.afterPropertiesSet();
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public boolean dispatch(Object event) {
        DisruptorQueue queue = queueMap.get(event.getClass());
        if (queue != null) {
            queue.push((EventData) event);
            return true;
        }
        return false;
    }

    @Override
    public void destroy() throws Exception {
        if (queues != null && queues.size() > 0) {
            executor.shutdown();
            queues.forEach(DisruptorQueue::stop);
        }
    }

    public ThreadPoolTaskExecutor getExecutor() {
        return executor;
    }
}
